package com.lzwk.app;

import com.lzwk.app.dwd.ocpx.DwdLookUpJoinTablesN;
import com.lzwk.app.function.GetJsonDouble;
import com.lzwk.app.function.GetJsonInt;
import com.lzwk.app.function.GetJsonString;
import com.lzwk.app.ods.adb.AdbCDCTables;
import com.lzwk.app.ods.adb.AdbJDBCTables;
import com.lzwk.app.ods.mdb.MdbJDBCTables;
import com.lzwk.app.ods.mdb.MdbLoookupTables;
import com.lzwk.app.ods.mdb.MdbLoookupTablesN;
import com.lzwk.app.ods.ocpx.OdsJDBCTables;
import com.lzwk.app.ods.wkForm.WkFormCDCTables;
import com.lzwk.app.ods.wkForm.WkFormJDBCTables;
import com.lzwk.base.BaseAppV1;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.StatementSet;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map;


/**
 * @title: CDC
 * @Author Yeaser
 * @Date: 2022/1/29 15:31
 * @Version 1.0
 * Description: TODO
 */
public class DBSSSum extends BaseAppV1 {

    public static void main(String[] args) {
        new DBSSSum().init(4,"cks","adb_dim_division","");
    }
    @Override
    protected void run(StreamExecutionEnvironment env, StreamTableEnvironment tenv) {



        tenv.getConfig().setIdleStateRetention(Duration.ofHours(3));//未更新的状态保留时间
        Configuration conf = tenv.getConfig().getConfiguration();

        conf.setString("table.exec.mini-batch.enabled","true"); //开启MiniBatch
        conf.setString("table.exec.mini-batch.allow-latency","20s"); //批量输出的间隔时间
        conf.setString("table.exec.mini-batch.size", "80000"); //防止OOM设置每个批次最多缓存数据的条数
        conf.setString("table.optimizer.agg-phase-strategy", "TWO_PHASE");//Global两阶段聚合缓解数据倾斜
        conf.setString("table.exec.async-lookup.buffer-capacity", "1000");//异步查找最大I/O数
        conf.setString("table.exec.async-lookup.timeout", "4min");//异步查找超时时间
        conf.setString("sql-client.verbose", "true");//详细error打印
        conf.setString("table.exec.window-agg.buffer-size-limit", "400000");//聚合窗口buffer大小


        tenv.getConfig().getConfiguration().setString("pipeline.name","adb_dim");

        //函数注册
        tenv.createTemporaryFunction("GetJsonString", GetJsonString.class);
        tenv.createTemporaryFunction("GetJsonInt", GetJsonInt.class);
        tenv.createTemporaryFunction("GetJsonDouble", GetJsonDouble.class);


        String sourceType = "source";
        String sinkType = "sink";
        String insertType = "insert";
        String queryType = "query";

        //JDBC表注册
        jdbcTables(tenv);

        //CDC表
        cdcTables(env,tenv);



        ArrayList<String> sinkSqls = joinTables().get(sinkType);
        for (String sql : sinkSqls) {

            tenv.executeSql(sql);

        }

        StatementSet statementSet = tenv.createStatementSet();

        ArrayList<String> insertSqls = joinTables().get(insertType);
        for (String insertSql : insertSqls) {

            statementSet.addInsertSql(insertSql);

        }

        try {

            statementSet.execute().wait();

        } catch (InterruptedException e) {

            e.printStackTrace();


        }


    }

    private void jdbcTables(StreamTableEnvironment tenv){
        //JDBC表注册
        AdbJDBCTables.adbaActivityTerm(tenv);
        AdbJDBCTables.adbAccountMapping(tenv);
        //AdbJDBCTables.adbClassStudentArchive(tenv);
        AdbJDBCTables.adbDivisionActvity(tenv);
        AdbJDBCTables.adbDivisionTermClass(tenv);
        MdbJDBCTables.mdbAccountAccount(tenv);
        //WkFormJDBCTables.wkFormAccountInfo(tenv);
        AdbJDBCTables.adbClassStudet(tenv);

    }


    private void cdcTables(StreamExecutionEnvironment env,StreamTableEnvironment tenv){
        AdbCDCTables.adbclassStudent(env,tenv);
        WkFormCDCTables.wkFormAccountInfo(env,tenv);
    }


    private Map<String, ArrayList<String>> joinTables(){

        String sourceType = "source";
        String sinkType = "sink";
        String insertType = "insert";
        HashMap<String, ArrayList<String>> res = new HashMap<>();

        ArrayList<String> sinkSqlList = new ArrayList<>();
        ArrayList<String> insertSqlList = new ArrayList<>();


        Map<String, String> classStudentSqls = MdbLoookupTablesN.adbDivisionClassStudent();

        sinkSqlList.add(classStudentSqls.get(sinkType));
        insertSqlList.add(classStudentSqls.get(insertType));

        Map<String, String> wkFormSqls = MdbLoookupTablesN.wkFormAccountInfo();

        sinkSqlList.add(wkFormSqls.get(sinkType));
        insertSqlList.add(wkFormSqls.get(insertType));

        res.put(sinkType,sinkSqlList);
        res.put(insertType,insertSqlList);

        return res;
    }
}
